package com.bdqn.spark.chapter06.bean

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark01_HotCategoryTop10Analysis {
  def main(args: Array[String]): Unit = {
    // TODO : Top10热门品类
    val sparConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("HotCategoryTop10Analysis")
    val sc = new SparkContext(sparConf)

    // todo：第一步：读取原始数据
//    val actionRDD: RDD[String] = sc.textFile("input/user_visit_action.csv")
    val actionRDD: RDD[String] = sc.textFile("input/word.txt")

    // TODO: 第二步：统计品类的点击总数：（品类，点击总数）
    val clickActionFilterRDD: RDD[String] = actionRDD.filter(
      // 2019-05-05,85,e2eef06e-beaa-4b49-acaf-38e057e1cd6e,36,2019-05-05 02:20:02,,-1,-1,,,1-2-3,1-2-3,8
      line => {
        val fields: Array[String] = line.split(",")
        fields(6) != "-1"
      }
    )

    val clickCategoryToOneRDD: RDD[(String, Int)] = clickActionFilterRDD.map(
      line => {
        val fields = line.split(",")
        (fields(6), 1)
      }
    )

    // 品类的点击总数 （品类，点击总数）
    val clickActionRDD: RDD[(String, Int)] =
      clickCategoryToOneRDD.reduceByKey(_ + _)

    // TODO: 第三步：统计品类的下单总数：（品类，下单总数）
    val orderActionFilterRDD: RDD[String] = actionRDD.filter(
      line => {
        val fileds: Array[String] = line.split(",")
        fileds(8) != ""
      }
    )

    // 一对多
    val orderCategoryToOneRDD: RDD[(String, Int)] = orderActionFilterRDD.flatMap(
      line => {
        val fields: Array[String] = line.split(",")
        val categoryIds: String = fields(8) // 1-2-3
        val arrIDs: Array[String] = categoryIds.split("-") // [1,2,3]
        arrIDs.map((_, 1)) // (1,1) (2,1) (3,1)
      }
    )

    // 品类的下单总数 （品类，下单总数）
    val orderActionRDD: RDD[(String, Int)] = orderCategoryToOneRDD.reduceByKey(_ + _)

    // TODO: 第四步：统计品类的支付总数：（品类，支付总数）
    val payActionFilterRDD: RDD[String] = actionRDD.filter(
      line => {
        val fileds: Array[String] = line.split(",")
        fileds(10) != ""
      }
    )

    // 一对多
    val payCategoryToOneRDD: RDD[(String, Int)] = payActionFilterRDD.flatMap(
      line => {
        val fields: Array[String] = line.split(",")
        val categoryIds: String = fields(10) // 1-2-3
        val arrIDs: Array[String] = categoryIds.split("-") // [1,2,3]
        arrIDs.map((_, 1)) // (1,1) (2,1) (3,1)
      }
    )

    // 品类的下单总数 （品类，下单总数）
    val payActionRDD: RDD[(String, Int)] = payCategoryToOneRDD.reduceByKey(_ + _)


    // TODO: （品类，点击总数） （品类，下单总数） （品类，支付总数）
    // TODO: （品类，（点击总数，下单总数，支付总数））
    // 一个是 把 三个rdd 连接在一起  connect
    // 另一个是 连完后，按照相同的品类，进行分组，group
    val cogroupRDD: RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Int]))] =
    clickActionRDD.cogroup(orderActionRDD, payActionRDD)

    val analysisRDD: RDD[(String, (Int, Int, Int))] = cogroupRDD.mapValues {
      case (clickIte, orderIte, payIte) => {
        var clickCnt = 0
        val iterClick = clickIte.iterator
        if (iterClick.hasNext){
          clickCnt = iterClick.next()
        }

        var orderCnt = 0
        val iterOrder: Iterator[Int] = orderIte.iterator
        if (iterOrder.hasNext){
          orderCnt = iterOrder.next()
        }

        var payCnt = 0
        val iterPay: Iterator[Int] = payIte.iterator
        if (iterPay.hasNext){
          payCnt = iterPay.next()
        }

        (clickCnt, orderCnt, payCnt)
      }
    }

    // 按照 元组的第二个元素（三元组）排序
    val result: Array[(String, (Int, Int, Int))] = analysisRDD.sortBy(_._2, false).take(10)
    result.foreach(println)

    sc.stop()
  }
}
